package task

import bean.Case.{CatergoryLogWide, Messages}
import org.apache.flink.streaming.api.scala._

/**
 *
 *   @program: util
 *   @author: YCLW058
 *   @create: 2021-05-27 15:35
 *   @decsription:
 *
 *   */

object DataToWideTask {
  def process(messagesDS: DataStream[Messages]): DataStream[CatergoryLogWide] = {
    messagesDS.map(msg => {
      CatergoryLogWide(
        msg.catergoryLog.categoryID,
        msg.catergoryLog.produceID,
        msg.catergoryLog.userID,
        msg.catergoryLog.price,
        msg.catergoryLog.country,
        msg.catergoryLog.network,
        msg.catergoryLog.source,
        msg.catergoryLog.buyType,
        msg.catergoryLog.buyTime,
        msg.catergoryLog.leaveTime,
        msg.count,
        msg.timeStamp
      )
    })
  }

}
